package com.sdhs.common.config;

import com.sdhs.common.utils.mqtt.MqttProperties;
import com.sdhs.common.utils.mqtt.service.MqttMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

/**
 * 收到订阅
 */
@Configuration
public class MqttInboundConfiguration {
	@Autowired
	private MqttProperties mqttProperties;
	@Autowired(required = false)
	private MqttPahoClientFactory mqttPahoClientFactory;
	@Autowired(required = false)
	MqttMQListener mqttMQListener;

	@Bean
	@ConditionalOnProperty(value = "mqtt.enabled", havingValue = "true", matchIfMissing = false)
	public MessageChannel mqttInputChannel() {
		return new DirectChannel();
	}

	@Bean
	@ConditionalOnProperty(value = "mqtt.enabled", havingValue = "true", matchIfMissing = false)
	public MessageProducer inbound() {
		// 获取所有topic注解,来订阅
		String inboundTopics = mqttProperties.getDefaultTopic() + "/#";
		MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
				mqttProperties.getClientId() + "c", mqttPahoClientFactory, inboundTopics);
		adapter.setCompletionTimeout(5000);
		adapter.setConverter(new DefaultPahoMessageConverter());
		adapter.setQos(1);
		adapter.setOutputChannel(mqttInputChannel());
		return adapter;
	}

	@Bean
	@ServiceActivator(inputChannel = "mqttInputChannel")
	public MessageHandler handler() {
		return new MessageHandler() {

			@Override
			public void handleMessage(Message<?> message) throws MessagingException {
				String topic = message.getHeaders().get("mqtt_receivedTopic").toString();
				String tag = topic.replaceFirst(mqttProperties.getDefaultTopic() + "/", "");
				//System.out.println("topic:"+ topic +"接受到消息--"+(String) message.getPayload());
				if (mqttMQListener != null) {
					mqttMQListener.onMessage(topic, tag, (String) message.getPayload());
				}
			}

		};
	}
}
